package edu.xhu.behavior.listener;

import cn.hutool.json.JSONUtil;
import edu.xhu.behavior.mapper.BiliUserChatDetailMapper;
import edu.xhu.behavior.mapper.BiliUserChatMapper;
import edu.xhu.common.exception.CustomException;
import edu.xhu.model.behavior.pojos.BiliUserChat;
import edu.xhu.model.behavior.pojos.BiliUserChatDetail;
import edu.xhu.model.common.constants.kafka.group.ChatConsumerGroup;
import edu.xhu.model.common.constants.kafka.topic.ChatTopic;
import edu.xhu.model.common.enums.AppHttpCodeEnum;
import edu.xhu.model.common.enums.IsFlagEnum;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.interceptor.TransactionAspectSupport;
import org.springframework.transaction.support.DefaultTransactionDefinition;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.stream.Collectors;

/**
 * IM处理消息类
 */
@Component
@RequiredArgsConstructor
@Slf4j
public class DetailSaveListener {

    private final BiliUserChatDetailMapper biliUserChatDetailMapper;

    private final BiliUserChatMapper biliUserChatMapper;

    private final KafkaTemplate<String, String> kafkaTemplate;

    private final Executor chatExecutor;

    private final DataSourceTransactionManager dataSourceTransactionManager;

    //批处理监听,异步批处理,减少打db次数
    @KafkaListener(topics = ChatTopic.BILL_CHAT_DETAIL_SAVE, groupId = ChatConsumerGroup.BILI_CHAT_GROUP)
    public void detail_save(List<String> messages,
                            @Header(KafkaHeaders.RECEIVED_PARTITION_ID) List<Integer> partitions,
                            @Header(KafkaHeaders.RECEIVED_TOPIC) List<String> topics) {
        log.info("开始持久化detail");
        long start = System.currentTimeMillis();
        if (messages != null && messages.size() > 0) {
            List<BiliUserChatDetail> allDetail = new ArrayList<>();
            //1.分离离线消息
            messages.forEach(str -> {
                BiliUserChatDetail detail = JSONUtil.toBean(str, BiliUserChatDetail.class);
                allDetail.add(detail);
            });
            if (!allDetail.isEmpty()) {
                //2.不管是否离线吧,异步持久化到db
                doRdb(allDetail);
                //3.对于离线消息,更新聊天表的未读数量(根据fromId和toId,批量更新)
                doUpdate(allDetail);
            }
        }
        long end = System.currentTimeMillis();
        log.info("持久化detail成功,耗时:{}ms", (end - start));
    }

    private void doUpdate(List<BiliUserChatDetail> allDetail) {
        CompletableFuture.runAsync(() -> {
            //手动开启事务
            DefaultTransactionDefinition def = new DefaultTransactionDefinition();
            // 事物隔离级别，开启新事务，这样会比较安全些。
            def.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
            // 获得事务状态
            TransactionStatus status = dataSourceTransactionManager.getTransaction(def);
            try {
                List<BiliUserChat> chats = allDetail.stream().map(detail -> {
                    BiliUserChat chat = new BiliUserChat();
                    BeanUtils.copyProperties(detail, chat);
                    return chat;
                }).collect(Collectors.toList());
                //批量更新(TODO 优化:把相同user->to,拎出来...(取舍,一次db循环n次,还是分类打n次db)...(磁盘IO与网络IO)...)
                if (biliUserChatMapper.updateBatchByFromAndToId(chats, 1) < 1) {
                    log.info("不知道为什么,批量更新聊天表未读消息失败");
                    throw new CustomException(AppHttpCodeEnum.SERVER_ERROR);
                }
            } catch (Exception e) {
                log.error("更新未读消息失败,错误信息:{}", e.toString());
                //事务回滚
                dataSourceTransactionManager.rollback(status);
                throw new CustomException(AppHttpCodeEnum.SERVER_ERROR);
            }
            //提交事务
            dataSourceTransactionManager.commit(status);
        }, chatExecutor);
    }

    private void doRdb(List<BiliUserChatDetail> allDetail) {
        CompletableFuture.runAsync(() -> {
            List<BiliUserChatDetail> all = new ArrayList<>(allDetail);
            //手动开启事务
            DefaultTransactionDefinition def = new DefaultTransactionDefinition();
            // 事物隔离级别，开启新事务，这样会比较安全些。
            def.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
            // 获得事务状态
            TransactionStatus status = dataSourceTransactionManager.getTransaction(def);
            try {
                //如果离线不为空且addAll失败
                if (all.isEmpty()) {
                    log.info("消息为空！！");
                    throw new CustomException(AppHttpCodeEnum.SERVER_ERROR);
                }
                if (biliUserChatDetailMapper.insertBatch(all) < 1) {
                    log.info("不知道为什么,批量插入失败");
                    throw new CustomException(AppHttpCodeEnum.SERVER_ERROR);
                }
                kafkaTemplate.send(ChatTopic.BILL_CHAT_RDB_PROCESS, JSONUtil.toJsonStr(all));
                log.info("给主题:{}发送了消息", ChatTopic.BILL_CHAT_RDB_PROCESS);
            } catch (Exception e) {
                log.error("持久化rdb或发送消息出错啦！！");
                //事务回滚
                dataSourceTransactionManager.rollback(status);
                throw new CustomException(AppHttpCodeEnum.SERVER_ERROR);
            }
            //提交事务
            dataSourceTransactionManager.commit(status);
        }, chatExecutor);
    }
}
